package com.asiainfo.dacp.datastash.broker.facade;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.StashUnitModel;
import com.asiainfo.dacp.datastash.broker.core.StashUnitProcessModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStateEnum;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitTaskStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitDefine;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitProcess;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTaskHeartbeat;
import com.asiainfo.dacp.datastash.broker.facade.service.UnitLoaderService;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitStateLogService;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitTaskStateLogService;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.domain.unit.repos.DataStashUnitRepository;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 *
 * unit与task的failover处理
 *
 * @author MeiKefu
 * @date 2018-1-2
 */
@Service
public class UnitFailoverFacade {

    @Autowired
    private StashTraceLogger tracelog;
    
    @Autowired
    private StashUnitModel stashUnitModel;
    
    @Autowired
    private StashUnitProcessModel stashUnitProcessModel;
    
    @Autowired
    private UnitStateLogService unitStateLogService;
    
    @Autowired
    private DataStashUnitRepository dataStashUnitRepository;
    
    @Autowired
    private UnitTaskStateLogService unitTaskStateLog;
    
    private String clzName = this.getClass().getSimpleName();



    public void unitFailover(){
        
    	List<StashUnitProcess> unitProcessList= stashUnitProcessModel.getUnitProcessList();
    	for (StashUnitProcess stashUnitProcess : unitProcessList) {
    		List<StashUnitTask> unitTasks = stashUnitProcess.getTasks();
    		
    		int succCount = 0;
    		int failCount = 0;
    		for (StashUnitTask stashUnitTask : unitTasks) {
				String taskState = stashUnitProcessModel.getUnitTaskState(stashUnitTask.getTaskProcId());
				if(taskState!=null&&taskState.equals(UnitTaskStateEnum.SUCESS.value())){
					succCount++;
					continue;
				}
				if(taskState!=null&&taskState.equals(UnitTaskStateEnum.FAIL.value())){
					failCount++;
					continue;
				}
			}
    		StashUnit stashUnit = stashUnitModel.getStashUnit(stashUnitProcess.getUnitProcId());
    		if(stashUnit != null){
    		if(stashUnit.getState().equals(UnitStateEnum.RUNNING.value())){
    			try {
        			if(succCount==unitTasks.size()){
        				stashUnitProcessModel.deleteUnitProcessList(stashUnitProcess.getUnitProcId());
            			stashUnit.success();
            			unitStateLogService.update(stashUnit,UnitStateEnum.SUCESS,"任务执行成功");
            			if(stashUnit.isTurnover()){
            				Map<String, StashUnitDefine> stashUnitDefs = BrokerContext.getInstance().getStashUnitDefs();
            				for (StashUnitDefine stashUnitDefine : stashUnitDefs.values()) {
            					if(stashUnitDefine.getUnitCode().equalsIgnoreCase(stashUnit.getUnitCode())) {
            						try {
            							String nextBatchNo = UnitLoaderService.lastBatchNo(stashUnit.getBatchNo(), stashUnitDefine.getBatchType(), stashUnitDefine.getBatchStep());
                	                    stashUnitModel.updateUnitBatchNo(stashUnitDefine.getUnitCode(), nextBatchNo);
                	                    dataStashUnitRepository.updateBatchNo(stashUnitDefine.getUnitCode(),nextBatchNo);
									} catch (Exception e) {
										e.printStackTrace();
									}
            					}
            				}
            				
            			}
            		}
            		if(failCount==unitTasks.size()){
            			stashUnitProcessModel.deleteUnitProcessList(stashUnitProcess.getUnitProcId());
            			stashUnit.failed();
            			unitStateLogService.update(stashUnit,UnitStateEnum.FAIL,"子任务全部失败,任务失败!");
            		}
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}else{
    			tracelog.info(new StashTraceLogVo(clzName,"unitFailover:error",stashUnitProcess.getUnitProcId()+":unit_process_list与unit_list队列的状态不一致"));
    		}
    	}
		}
    	
    }

    public void taskFailover(){
        //1.每5分钟监听，扫描unit_task_state_list，检查unit_task_heartbeat_list是否有心跳
        //2.重新放入unit_task_wait_queue或unit_task_dispatch_queue队列
    	Map<String,String> taskStateMap = stashUnitProcessModel.getUnitTaskState();
    	Iterator<Map.Entry<String, String>> taskStareIt = taskStateMap.entrySet().iterator();
    	while (taskStareIt.hasNext()) {
			Map.Entry<String, String> entry =  taskStareIt.next();
			if(UnitTaskStateEnum.RUNNING.value().equals(entry.getValue())){
				StashUnitTaskHeartbeat taskHeartBeat =stashUnitProcessModel.getTaskHeartBeat(entry.getKey());
				if(taskHeartBeat!=null){
					if(!checkHeartBeat(taskHeartBeat.getTimestamp())){
						StashUnitTask stashUnitTask = getStashUnitTask(taskHeartBeat.getUnitProcId(), taskHeartBeat.getTaskProcId());
						if(stashUnitTask!=null){
							
							stashUnitProcessModel.pushUnitTaskDispatchQueue(stashUnitTask);
							unitTaskStateLog.update(stashUnitTask.getTaskProcId(),stashUnitTask.getUnitProcId(),stashUnitTask.getUnitCode(),stashUnitTask.getBatchNo(),
					                UnitTaskStateEnum.BLOCKED_RESOURCE.value(),"work失联,task重做",null,null
					        );
							//根据pid查杀进程暂未实现
						}else{
							tracelog.info(new StashTraceLogVo(clzName,"taskFailover:error",entry.getKey()+":获取不到任务信息"));
						}
						
					}
				}else{
					tracelog.info(new StashTraceLogVo(clzName,"taskFailover:warn",entry.getKey()+":没有心跳信息"));
				}
			}
		}
    }
    
    public StashUnitTask getStashUnitTask(String unitProcId,String taskProcId){
    	StashUnitTask stashUnitTask=null;
    	StashUnitProcess stashUnitProcess = stashUnitProcessModel.getUnitProcessList(unitProcId);
    	for (StashUnitTask unitTask : stashUnitProcess.getTasks()) {
			if(taskProcId.equals(unitTask.getTaskProcId())){
				stashUnitTask=unitTask;
				break;
			}
		}
    	
    	return stashUnitTask;
    }
   
    public boolean checkHeartBeat(Long timestamp){
    	boolean flag = true;
    	long currentTime = System.currentTimeMillis();
    	
    	if(currentTime-timestamp>30){
    		flag=false;
    	}
    	return flag;
    }
}
